package cn.hy.dataSync.statistics.communication;

import com.alibaba.fastjson.JSON;
import org.apache.commons.lang.Validate;

import java.text.DecimalFormat;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;


public final class CommunicationTool {
    // ？
    public static final String STAGE = "stage";
    // 读写记录的速度
    public static final String RECORD_SPEED = "recordSpeed";
    // 成功的百分比
    public static final String PERCENTAGE = "percentage";

    public static final String READ_SUCCEED_RECORDS = "readSucceedRecords";

    // 读失败记录-key
    public static final String READ_FAILED_RECORDS = "readFailedRecords";

    public static final String WRITE_RECEIVED_RECORDS = "writeReceivedRecords";
    // 写失败记录-key
    public static final String WRITE_FAILED_RECORDS = "writeFailedRecords";

    public static final String TOTAL_READ_RECORDS = "totalReadRecords";

    // 总的记录错误信息
    private static final String TOTAL_ERROR_RECORDS = "totalErrorRecords";

    private static final String WRITE_SUCCEED_RECORDS = "writeSucceedRecords";

    public static final String WAIT_WRITER_TIME = "waitWriterTime";

    public static final String WAIT_READER_TIME = "waitReaderTime";

    // 字段转换所用的时间
    public static final String TRANSFORMER_USED_TIME = "totalTransformerUsedTime";
    public static final String TRANSFORMER_SUCCEED_RECORDS = "totalTransformerSuccessRecords";
    public static final String TRANSFORMER_FAILED_RECORDS = "totalTransformerFailedRecords";
    public static final String TRANSFORMER_FILTER_RECORDS = "totalTransformerFilterRecords";

    /**
     * 用来统计数据
     * @param now  Communication
     * @param old Communication  用于获取时间戳和异常
     * @param totalStage totalStage
     * @return
     */
    public static Communication getReportCommunication(Communication now, Communication old, int totalStage) {
        Validate.noNullElements(new Communication[]{now, old}, "为汇报准备的新旧 communication 不能为 null");

        long totalReadRecords = getTotalReadRecords(now);
        now.setLongCounter(TOTAL_READ_RECORDS, totalReadRecords);
        now.setLongCounter(TOTAL_ERROR_RECORDS, getTotalErrorRecords(now));
        now.setLongCounter(WRITE_SUCCEED_RECORDS, getWriteSucceedRecords(now));

        long timeInterval = now.getTimestamp() - old.getTimestamp();
        long sec = timeInterval <= 1000 ? 1 : timeInterval / 1000;
        long recordsSpeed = (totalReadRecords - getTotalReadRecords(old)) / sec;

        now.setLongCounter(RECORD_SPEED, recordsSpeed < 0 ? 0 : recordsSpeed);
        now.setDoubleCounter(PERCENTAGE, now.getLongCounter(STAGE) / (double) totalStage);

        if (old.getThrowable() != null) {
            now.setThrowable(old.getThrowable());
        }
        return now;
    }

    public static long getTotalReadRecords(final Communication communication) {
        return communication.getLongCounter(READ_SUCCEED_RECORDS) +
                communication.getLongCounter(READ_FAILED_RECORDS);
    }

    public static long getTotalErrorRecords(final Communication communication) {
        return communication.getLongCounter(READ_FAILED_RECORDS) +
                communication.getLongCounter(WRITE_FAILED_RECORDS);
    }

    public static long getWriteSucceedRecords(final Communication communication) {
        return communication.getLongCounter(WRITE_RECEIVED_RECORDS) -
                communication.getLongCounter(WRITE_FAILED_RECORDS);
    }

    public static class Stringify {
        private final static DecimalFormat DECIMAL_FORMAT = new DecimalFormat("0.00");

        public static String getSnapshot(final Communication communication) {
            StringBuilder sb = new StringBuilder();
            sb.append("Total "); // 总记录数
            sb.append(getTotal(communication));
            sb.append(" | ");
            sb.append("Speed ");// 速度
            sb.append(getSpeed(communication));
            sb.append(" | ");
            sb.append("Error "); // 错误信息
            sb.append(getError(communication));
            sb.append(" | ");
            sb.append(" All Task WaitWriterTime "); // 等待时间 纳秒
            sb.append(unitTime(communication.getLongCounter(CommunicationTool.WAIT_WRITER_TIME)));
            sb.append(" | ");
            sb.append(" All Task WaitReaderTime ");
            sb.append(unitTime(communication.getLongCounter(CommunicationTool.WAIT_READER_TIME)));
            sb.append(" | ");
            if (communication.getLongCounter(CommunicationTool.TRANSFORMER_USED_TIME) > 0
                    || communication.getLongCounter(CommunicationTool.TRANSFORMER_SUCCEED_RECORDS) > 0
                    || communication.getLongCounter(CommunicationTool.TRANSFORMER_FAILED_RECORDS) > 0
                    || communication.getLongCounter(CommunicationTool.TRANSFORMER_FILTER_RECORDS) > 0) {
                sb.append("Transformer Success ");
                sb.append(String.format("%d records", communication.getLongCounter(CommunicationTool.TRANSFORMER_SUCCEED_RECORDS)));
                sb.append(" | ");
                sb.append("Transformer Error ");
                sb.append(String.format("%d records", communication.getLongCounter(CommunicationTool.TRANSFORMER_FAILED_RECORDS)));
                sb.append(" | ");
                sb.append("Transformer Filter ");
                sb.append(String.format("%d records", communication.getLongCounter(CommunicationTool.TRANSFORMER_FILTER_RECORDS)));
                sb.append(" | ");
                sb.append("Transformer usedTime ");
                sb.append(unitTime(communication.getLongCounter(CommunicationTool.TRANSFORMER_USED_TIME)));
                sb.append(" | ");
            }
            sb.append("Percentage "); // 成功率
            sb.append(getPercentage(communication));
            return sb.toString();
        }

        private static String getTotal(final Communication communication) {
            return String.format("%d records", communication.getLongCounter(TOTAL_READ_RECORDS));
        }

        private static String getError(final Communication communication) {
            return String.format("%d records", communication.getLongCounter(TOTAL_ERROR_RECORDS));
        }

        private static String getSpeed(final Communication communication) {
            return String.format("%d records/s", communication.getLongCounter(RECORD_SPEED));
        }

        private static String getPercentage(final Communication communication) {
            return DECIMAL_FORMAT.format(communication.getDoubleCounter(PERCENTAGE) * 100) + "%";
        }
    }

    public static class Jsonify {
        @SuppressWarnings("rawtypes")
        public static String getSnapshot(Communication communication) {
            Validate.notNull(communication);

            Map<String, Object> state = new HashMap<>();

            Pair pair = getTotalRecords(communication);
            state.put((String) pair.getKey(), pair.getValue());

            pair = getSpeedRecord(communication);
            state.put((String) pair.getKey(), pair.getValue());

            pair = getStage(communication);
            state.put((String) pair.getKey(), pair.getValue());

            pair = getErrorRecords(communication);
            state.put((String) pair.getKey(), pair.getValue());

            pair = getErrorMessage(communication);
            state.put((String) pair.getKey(), pair.getValue());

            pair = getPercentage(communication);
            state.put((String) pair.getKey(), pair.getValue());

            pair = getWaitReaderTime(communication);
            state.put((String) pair.getKey(), pair.getValue());

            pair = getWaitWriterTime(communication);
            state.put((String) pair.getKey(), pair.getValue());

            return JSON.toJSONString(state);
        }

        private static Pair<String, Long> getTotalRecords(final Communication communication) {
            return new Pair<>("totalRecords", communication.getLongCounter(TOTAL_READ_RECORDS));
        }

        private static Pair<String, Long> getSpeedRecord(final Communication communication) {
            return new Pair<>("speedRecords", communication.getLongCounter(RECORD_SPEED));
        }

        private static Pair<String, Long> getErrorRecords(final Communication communication) {
            return new Pair<>("errorRecords", communication.getLongCounter(TOTAL_ERROR_RECORDS));
        }

        private static Pair<String, Long> getStage(final Communication communication) {
            return new Pair<>("stage", communication.getLongCounter(STAGE));
        }

        private static Pair<String, Double> getPercentage(final Communication communication) {
            return new Pair<>("percentage", communication.getDoubleCounter(PERCENTAGE));
        }

        private static Pair<String, String> getErrorMessage(final Communication communication) {
            return new Pair<>("errorMessage", communication.getThrowableMessage());
        }

        private static Pair<String, Long> getWaitReaderTime(final Communication communication) {
            return new Pair<>("waitReaderTime", communication.getLongCounter(CommunicationTool.WAIT_READER_TIME));
        }

        private static Pair<String, Long> getWaitWriterTime(final Communication communication) {
            return new Pair<>("waitWriterTime", communication.getLongCounter(CommunicationTool.WAIT_WRITER_TIME));
        }

        static class Pair<K, V> {
            public Pair(final K key, final V value) {
                this.key = key;
                this.value = value;
            }

            public K getKey() {
                return key;
            }

            public V getValue() {
                return value;
            }

            private K key;

            private V value;
        }
    }

    private static String unitTime(long time) {
        return unitTime(time, TimeUnit.NANOSECONDS);
    }

    private static String unitTime(long time, TimeUnit timeUnit) {
        return String.format("%,.3fs", ((float) timeUnit.toNanos(time)) / 1000000000);
    }
}
